package com.chj.nettydemo.direct.handler;


import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.EventExecutorGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Component
@ChannelHandler.Sharable
public class Gb32ServerHandler extends SimpleChannelInboundHandler<String> {

    public static final Map<ChannelHandlerContext, String> CHANNEL_VIN = new ConcurrentHashMap<>();

    public static final Map<String, ChannelHandlerContext> VIN_CHANNEL = new ConcurrentHashMap<>();

    public static final ChannelGroup CHANNEL_GROUP = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    public static final EventExecutorGroup EVENT_EXECUTOR_GROUP = new DefaultEventExecutorGroup(Runtime.getRuntime().availableProcessors() * 2, new DefaultThreadFactory("async-workers", true));

    // 一旦连接，第一个执行
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        // 将当前 channel 加入 ChannelGroup
        CHANNEL_GROUP.add(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        // 将当前 channel 移出 ChannelGroup
        log.info("客户端断开连接【{}】", ctx.channel().remoteAddress());
        // handlerRemoved 触发后 自动移除
        CHANNEL_GROUP.remove(ctx.channel());
        String uniqId = CHANNEL_VIN.get(ctx);
        if (!StringUtils.hasLength(uniqId)) {
            VIN_CHANNEL.remove(uniqId);
        }
        // CHANNEL_VIN 统一移除处理 VIN_CHANNEL 需在各个环节结合业务情况移除
        CHANNEL_VIN.remove(ctx);
    }

    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端已上线【{}】 【{}】", ctx.channel().remoteAddress(), this.hashCode());
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端正常下线【{}】", ctx.channel().remoteAddress());
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, String msg) throws Exception {
        // 异步处理
        this.asyncExec(msg, ctx);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        cause.printStackTrace();
        log.error("【{}】,客户端异常下线,【{}】", cause.getMessage(), ctx.channel().remoteAddress());
        ctx.pipeline().remove(this);
        ctx.close();
    }

    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (IdleState.ALL_IDLE == event.state()) {
                log.warn(ctx.channel().remoteAddress() + "--读写超时--" + event.state());
                String uniqId = CHANNEL_VIN.get(ctx);
                if (!StringUtils.hasLength(uniqId)) {
                    VIN_CHANNEL.remove(uniqId);
                }
                ctx.close();
            }
        }
    }

    private void asyncExec(String msg, ChannelHandlerContext ctx) {
        EVENT_EXECUTOR_GROUP.submit(() -> {
            log.info("==客户端发送报文=={}", msg);
            // 业务处理流程 TODO
            ctx.writeAndFlush("hello client");
        });
    }
}